package Autodesk;

import java.io.IOException;
import java.util.Hashtable;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.net.URLCodec;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class TotalQueryFilter extends Configured implements Tool  {
	
		public Path prodPath;

		public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
			
			private final static IntWritable one = new IntWritable(1);
			private Text word = new Text();
			Hashtable<String, String> ht = new Hashtable<String, String>();

		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			StringBuffer bf = new StringBuffer();
			String queryParsed="2";
			String record[] = value.toString().toLowerCase().split("\t", 11);
			try{
				bf.append(record[0]);// Date.
				bf.append("\t");
				
				bf.append(record[1]);// Channel			
				bf.append("\t");
				
				bf.append(record[4]);// Language
				bf.append("\t");
				
				bf.append(record[3]);// Product			
				bf.append("\t");
				
				String query=record[6];// Query
				byte[] buf=query.getBytes("UTF-8");
				byte[] buf2=URLCodec.decodeUrl(buf);
				queryParsed = new String(buf2);	
				
				queryParsed = eliminateTabsQuery(queryParsed);
				bf.append(queryParsed);

				bf.append("\t");
				bf.append(record[8]);// Country
				
				word.set(bf.toString());
				
				Pattern queryPattern = Pattern.compile("\\S");//VALIDATES THAT THE QUERIES ARE NOT SPACES
				Matcher queryMatcher = queryPattern.matcher(queryParsed);
				if( !queryParsed.equals("2") && !queryParsed.isEmpty() && queryMatcher.find() ){
					context.write(word, one);
				}
			}catch(Exception e){
				System.out.println("Erroor Reg: "+value);
			}
		}

	}
		public static String eliminateTabsQuery(String queryParsed) {
			Pattern tabPattern = Pattern.compile("\t");	//Eliminate tabs
			Matcher tabMatcher = tabPattern.matcher(queryParsed);
			if(tabMatcher.find()){
				queryParsed=queryParsed.replaceAll("\t", " ");
			}
			return queryParsed;
		}

	public static class LogReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		
		private IntWritable result = new IntWritable(1);
		private MultipleOutputs<Text,IntWritable> mos;
		 
		 public void setup(Context context) {
			 mos = new MultipleOutputs<Text,IntWritable>(context);
		 }
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			if(sum!=0){
				result.set(sum);
			}
			//context.write(key, result);
			String[] keySplitted=key.toString().split("\t",2);
			mos.write(key, result, keySplitted[0]+"_QueryReport");

		}
		 public void cleanup(Context context) throws IOException, InterruptedException {
			 mos.close();
		 }
	}
	
	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "QueryReport");
		job.setJarByClass(Autodesk.TotalQueryFilter.class);

		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

	    FileInputFormat.setInputPaths(job, new Path(args[0]));
	    FileOutputFormat.setOutputPath(job, new Path(args[1]) );
	    MultipleOutputs.addNamedOutput(job,"QueryReport", TextOutputFormat.class, Text.class, IntWritable.class);

		System.exit(job.waitForCompletion(true)?0:1);
		
		return 0;
	
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new TotalQueryFilter(), args);
		System.exit(res);
		}

 
}
